package com.yinanxi.cool

import java.util.Properties

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.SparkSession

/**
 * @Created by Orion
 * @Description
 * 根据原始地理位置信息  构建便于查询的地理位置字典表
 */
object GenAreaDict {

  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder()
      .master("local[*]")
      .appName("session-split")
      .enableHiveSupport()
      .getOrCreate()

    val properties = new Properties()
    properties.setProperty("user", "root")
    properties.setProperty("password", "root")
    // 读取MySQL中的数据
    val df = session.read.jdbc("jdbc:mysql://localhost:3306/cool?useUnicode=true&characterEncoding=utf8", "t_md_areas", properties)
    // 创建临时表
    df.createOrReplaceTempView("tb_area")
    // 自定义函数
    def mygeo(lat: Double, lnt: Double): String = {
      // GeoHash.geoHashStringWithCharacterPrecision(lat ,lnt , 6)
      val geoStr = GeoHash.withCharacterPrecision(lat, lnt, 6).toBase32
      geoStr
    }
    /*   val f = (lat:Double , lnt:Double)=>{
         val geoStr = GeoHash.withCharacterPrecision(lat, lnt, 6).toBase32
         geoStr
       }*/
    // 注册函数
    session.udf.register("mygeo", mygeo _)

    // 处理
    val resDF = session.sql(
      """
        |SELECT
        | mygeo(x3.BD09_LAT, x3.BD09_LNG)  as  geostr ,
        |x1.AREANAME as  province ,
        |x2.AREANAME as city ,
        |x3.AREANAME as region ,
        |"" as street ,
        |x3.BD09_LAT as  lat ,
        |x3.BD09_LNG as  lng
        |FROM
        |tb_area x3
        |join
        |tb_area x2
        |JOIN
        |tb_area x1
        |on
        |x3.PARENTID = x2.ID
        |and
        |x2.PARENTID = x1.ID
        |and  x3.`LEVEL` = '3'
        |union
        |SELECT
        | mygeo(x4.BD09_LAT, x4.BD09_LNG)  geostr ,
        |x1.AREANAME as  province ,
        |x2.AREANAME as city ,
        |x3.AREANAME as region ,
        |x4.AREANAME as street ,
        |x4.BD09_LAT as  lat ,
        |x4.BD09_LNG as  lng
        |FROM
        |tb_area x4
        |join
        |tb_area x3
        |JOIN
        |tb_area x2
        |JOIN
        |tb_area x1
        |on
        |x4.PARENTID = x3.ID
        |and
        |x3.PARENTID = x2.ID
        |and
        |x2.PARENTID = x1.ID
        |and  x4.`LEVEL` = '4'
        |""".stripMargin)

    // 保存 将数据存储在mysql中
    resDF.write.jdbc("jdbc:mysql://localhost:3306/cool?useUnicode=true&characterEncoding=utf8", "tb_area_dict", properties)

    session.close()

  }

}
